package ranks.tasks.get_ready;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;

import mapred.CombineConfig;
import mapred.InputConfig;
import mapred.Job;
import mapred.MapConfig;
import mapred.OutputConfig;
import mapred.ReduceConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.TextOutputFormat;

public class GetReady extends Job {

	private String output;

	public GetReady(String... settings) {
		super("Get Ready", settings);
		this.output = this.getConfig("-output");

		this.setConfig("-reduce", 0);
	}

	@Override
	protected CombineConfig getCombineConfig() {
		return CombineConfig.newInstance(null);
	}

	@Override
	protected InputConfig getInputConfig(FileSystem fs) throws IOException {
		return InputConfig.newInstance(KeyValueTextInputFormat.class, this
				.getConfig("-linkgraph"));
	}

	@Override
	protected MapConfig getMapConfig() {
		return MapConfig.newInstance(GetReady_Mapper.class, 0, Text.class,
				Text.class, DefaultCodec.class);
	}

	@Override
	protected OutputConfig getOutputConfig(FileSystem fs) throws IOException {
		return OutputConfig.newInstance(TextOutputFormat.class, Text.class,
				Text.class, output + "/Round_0",
				GzipCodec.class);
	}

	@Override
	protected ReduceConfig getReduceConfig() {
		return ReduceConfig.newInstance(null, 0);
	}

	@Override
	protected void setUp(JobConf conf, FileSystem fs) throws IOException {
        this.setJobConfConfig("Output", output);
        this.setJobConfConfig("Round", 0);
        
        this.setJobConfConfig("Seeds", this.getConfig("-seeds"));

        BufferedReader br = new BufferedReader(new InputStreamReader(fs
                .open(new Path(this.getConfig("-seeds") + ".times"))));
        this.setJobConfConfig("SeedsTimes", br.readLine().trim());
        br.close();
	}

	@Override
	protected void tearDown(JobConf conf, FileSystem fs, RunningJob job)
			throws IOException {
		long websize = this.getCounter(job, "WebSize", "WebSize");
		
		BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fs
				.create(new Path(output + "/WebSize"), true)));
		bw.write(String.valueOf(websize));
		bw.flush();
		bw.close();
		
		fs.delete(new Path(output + "/Z_0"), true);
		fs.delete(new Path(this.getConfig("-seeds") + ".times"), true);
	}

}
